
package cn.tang.tframe.mq.kafka.consumer;

import cn.tang.tframe.mq.IQtMessager;
import cn.tang.tframe.mq.consumer.IQtMsgProcessor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.Serializable;
import java.util.Arrays;

/**
 * Created by tanghc on 17/2/4.
 */
public class DelegateKafkaConsumer<T extends Serializable>
	implements Runnable {

	private String topic;
	private KafkaConsumer<String, IQtMessager<T>> consumer;
	private IQtMsgProcessor msgProcessor;

	public DelegateKafkaConsumer(String topic, KafkaConsumer<String, IQtMessager<T>> consumer, IQtMsgProcessor msgProcessor) {
		this.topic = topic;
		this.consumer = consumer;
		this.msgProcessor = msgProcessor;
	}

	@Override
	public void run() {
		if(null == consumer || msgProcessor == null){
			return ;
		}
		consumer.subscribe(Arrays.asList(topic));
		while (true){
			ConsumerRecords<String, IQtMessager<T>> records = consumer.poll(100);
			if(records != null){
				for (ConsumerRecord<String, IQtMessager<T>> record : records){
					if(null != record){
						IQtMessager<T> msg = record.value();
						if(null != msg){
							msgProcessor.onConsume(msg.getMsgId(),msg.getBody());
						}
					}
				}
			}
		}
	}
}
